package com.yanggu.flink.datastream_api.source

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig

/**
 * 从rabbitmq中获取数据
 * flink官方网站
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/datastream/rabbitmq/
 */
object SourceFromRabbitMQ {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val tool = ParameterTool.fromPropertiesFile(getClass.getResourceAsStream("/rabbitmq.properties"))

    //配置rabbitmq的连接信息
    val rabbitmqConnectConfig = new RMQConnectionConfig.Builder()
      .setHost(tool.get("rabbitmq.host"))
      .setPort(tool.getInt("rabbitmq.port"))
      .setVirtualHost(tool.get("rabbitmq.virtualHost"))
      .setUserName(tool.get("rabbitmq.username"))
      .setPassword(tool.get("rabbitmq.password"))
      .build()

    environment
      .addSource(new RMQSource[String](rabbitmqConnectConfig, tool.get("rabbitmq.source.queue"), new SimpleStringSchema))
      .print()

    environment.execute("SourceFromRabbitMQ Job")

  }

}
